Skip to content

[SPARK-61180][CONNECT][PYTHON] Add DataFrame._joinAsOf, the as-of join of pandas-on-Spark's merge_asof, in the Rust and Python clients as PySpark does - #376

Open
HyukjinKwon wants to merge 1 commit into
apache:masterfrom
HyukjinKwon:SPARK-61180-join-as-of
Open

HyukjinKwon wants to merge 1 commit into
apache:masterfrom
HyukjinKwon:SPARK-61180-join-as-of

Conversation

@HyukjinKwon

Copy link
Copy Markdown
Member

What changes were proposed in this pull request?

SPARK-61180: PySpark Connect's DataFrame._joinAsOf(other, leftAsOfColumn, rightAsOfColumn, on=None, how=None, *, tolerance=None, allowExactMatches=True, direction="backward") builds the Spark Connect AsOfJoin relation. pandas-on-Spark's merge_asof calls it.

  • Rust core: DataFrame::as_of_join(&right, AsOfJoinOptions) with a new AsOfJoinOptions struct (AsOfJoinOptions::new(left_as_of, right_as_of) sets the PySpark defaults: inner, exact matches allowed, "backward"). The plan is a new LogicalPlan::AsOfJoin, serialized to AsOfJoin with left_as_of, right_as_of, join_expr or using_columns, join_type, tolerance, allow_exact_matches and direction. A DataFrame of another session fails with SESSION_NOT_SAME.
  • PyO3: DataFrame._joinAsOf with PySpark's signature. how defaults to inner and must be a string. tolerance must be a Column. on is a name, a Column, or a list of names or Columns, as in plan.AsOfJoin. A string as-of column is a column of its DataFrame, as df._col(name). join_type and direction are sent as they are, so the server checks them as it does for PySpark.

Why are the changes needed?

ps.merge_asof failed, because _joinAsOf did not exist and the name resolved through DataFrame.__getattr__ to a Column:

>>> ps.merge_asof(ps.DataFrame(L), ps.DataFrame(R), on="t")
TypeError: 'pyspark.sql.connect.column.Column' object is not callable

PySpark returns [[1, 1.0, 'a', nan, None], [5, 5.0, 'b', 4.0, 'y']] for L = [(1, 1.0, "a"), (5, 5.0, "b")] ("t int, x double, k string") and R = [(2, 2.0, "x"), (4, 4.0, "y")] ("t int, y double, k string").

Does this PR introduce any user-facing change?

Yes. DataFrame._joinAsOf and pandas.merge_asof (pandas-on-Spark) work on the Rust client as they do on PySpark Connect. Before, they raised the TypeError above.

How was this patch tested?

  • test_join_as_of_sends_the_as_of_join_as_pyspark (offline, python/tests/test_dropin_offline.py): checks the fields of the sent AsOfJoin relation for a list on with how, tolerance, allowExactMatches=False and direction, for the defaults, and for a Column on.
  • test_join_as_of_and_merge_asof_give_the_rows_of_pyspark (e2e, python/tests/test_dropin_e2e.py): the rows of _joinAsOf for each direction, allowExactMatches, tolerance, on as a list, and the inner default, and the rows of ps.merge_asof with on, by, direction, tolerance, and left_on/right_on/left_by/right_by.
  • as_of_join_sends_every_field_of_the_as_of_join, as_of_join_defaults_and_sends_a_condition_as_join_expr and as_of_join_rejects_a_dataframe_of_another_session (Rust unit tests, crates/spark-connect/src/dataframe.rs).
  • The offline test fails without this change, with TypeError: 'pyspark.sql.connect.column.Column' object is not callable. The same cases give the same schemas, rows, and error messages on official PySpark 4.3.0 against the same server. The explain text matches apart from plan IDs.
  • The lib tests of both crates (611 passed), the build of all tests, also with wasm-udf, the offline Python suites (2228 passed, 8 skipped), and the live drop-in and Arrow suites (211 passed) and live cases (10 passed) against a local Spark 4.3.0 server. Clippy reports no diagnostics in the changed lines.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Opus 5.5

This pull request and its description were written by Isaac.

…n of pandas-on-Spark's merge_asof, in the Rust and Python clients as PySpark does

Add DataFrame._joinAsOf to the Rust core, the PyO3 drop-in and its tests. PySpark Connect
builds the AsOfJoin relation for it, and pandas-on-Spark's merge_asof calls it. The as-of
join takes the as-of columns, the join condition or the shared column names, the join type,
an optional tolerance, allowExactMatches, and the direction, and the server runs it.

Co-authored-by: Isaac <no-reply@databricks.com>

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant